package com.nebulagraph.increment;

import java.io.File;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

import com.alibaba.fastjson.parser.deserializer.BigDecimalDeserializer;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.client.*;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.vesoft.nebula.client.graph.NebulaPoolConfig;
import com.vesoft.nebula.client.graph.data.HostAddress;
import com.vesoft.nebula.client.graph.data.ResultSet;
import com.vesoft.nebula.client.graph.data.ValueWrapper;
import com.vesoft.nebula.client.graph.exception.AuthFailedException;
import com.vesoft.nebula.client.graph.exception.IOErrorException;
import com.vesoft.nebula.client.graph.exception.NotValidConnectionException;
import com.vesoft.nebula.client.graph.net.NebulaPool;
import com.vesoft.nebula.client.graph.net.Session;
import org.apache.commons.cli.*;
import org.apache.commons.lang.StringUtils;

public class Mysql {
    private Config config;
    private String database;
    public Map<String, Config> tagConfigMap;
    public Map<String, Config> edgeConfigMap;
    public Map<String, List<Map<String, String>>> propMap;
    public Map<String, List<Map<String, String>>> tagPropMap;
    public Map<String, List<Map<String, String>>> edgePropMap;
    public Mysql(File cfgFile) {
        this.config = ConfigFactory.parseFile(cfgFile);

        List<Config> tags = (List<Config>) this.config.getConfigList("tags");
        List<Config> edges = (List<Config>) this.config.getConfigList("edges");
        this.tagConfigMap = new HashMap<String, Config>();
        this.edgeConfigMap = new HashMap<String, Config>();
        this.tagPropMap = new HashMap<String, List<Map<String, String>>>();
        this.edgePropMap = new HashMap<String, List<Map<String, String>>>();
        this.propMap = new HashMap<String, List<Map<String, String>>>();

        for (Config tag : tags) {
            this.tagConfigMap.put(tag.getString("table"), tag);
            Config n = tag.getConfig("nebula");
            Config vertex = tag.getConfig("vertex");
            Map<String, String> prop = new HashMap<String, String>();
            List<Map<String, String>> listProp = new ArrayList<Map<String, String>>();
            List<String> mysqlFields = tag.getStringList("fields");
            List<String> nebulaFields = n.getStringList("fields");
            prop.put("NEBULA_TAG_NAME", tag.getString("name"));
            prop.put(vertex.getString("field"), "vertex");
            for (int i = 0; i < nebulaFields.size(); i++) {
                prop.put(mysqlFields.get(i), nebulaFields.get(i));
            }

            List<Map<String, String>> preMap = null;
            if (this.tagPropMap.containsKey(tag.getString("table"))) {
                preMap = this.tagPropMap.get(tag.getString("table"));
            } else {
                preMap = new ArrayList<Map<String, String>>();
            }
            preMap.add(prop);
            this.tagPropMap.put(tag.getString("table"), preMap);
            this.propMap.put(tag.getString("table"), preMap);
        }
        for (Config edge : edges) {
            this.edgeConfigMap.put(edge.getString("table"), edge);
            Config n = edge.getConfig("nebula");
            Map<String, String> prop = new HashMap<String, String>();
            List<String> mysqlFields = edge.getStringList("fields");
            List<String> nebulaFields = n.getStringList("fields");
            prop.put("NEBULA_EDGE_NAME", edge.getString("name"));
            prop.put("source", edge.getConfig("source").getString("field"));
            prop.put("target", edge.getConfig("target").getString("field"));
            for (int i = 0; i < nebulaFields.size(); i++) {
                prop.put(mysqlFields.get(i), nebulaFields.get(i));
            }


            List<Map<String, String>> preMap = null;
            if (this.edgePropMap.containsKey(edge.getString("table"))) {
                preMap = this.edgePropMap.get(edge.getString("table"));
            } else {
                preMap = new ArrayList<Map<String, String>>();
            }
            preMap.add(prop);
            this.edgePropMap.put(edge.getString("table"), preMap);
            this.propMap.put(edge.getString("table"), preMap);
        }
    }

    public void syn(String host, Integer port, String destination, String username, String password, String database) {
        this.database = database;
        int batchSize = 1;
        while (true) {
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(host, port), destination, username, password);
            try {
                connector.connect();
                connector.subscribe(database + "\\..*");
                connector.rollback();
                System.out.println("start successful...");
                while (true) {
                    Message message = connector.getWithoutAck(batchSize);
                    long batchId = message.getId();
                    int size = message.getEntries().size();
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    } else {
                        parseEntry(message.getEntries());
                    }
                    connector.ack(batchId);
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                connector.disconnect();
            }
        }
    }

    public static void main(String args[]) throws ParseException {
        CommandLineParser parser = new BasicParser( );
        Options options = new Options( );

        String[] ops = new String[]{"config","host","port","destination","username","password","database"};
        for (String op : ops) {
            options.addOption(op, op, true, "config " + op + ".");
        }
        CommandLine commandLine = parser.parse( options, args );
        for (String op : ops) {
            if (!commandLine.hasOption(op)) {
                System.out.println("please input config " + op + "!!!");
                return;
            }
        }

        File cfgFile = null;
        cfgFile = new File(commandLine.getOptionValue("config"));
        if (!cfgFile.exists()) {
            System.out.println("the file not found!!!");
            return;
        }
        String host = commandLine.getOptionValue("host");
        String port = commandLine.getOptionValue("port");
        String destination = commandLine.getOptionValue("destination");
        String username = commandLine.getOptionValue("username");
        String password = commandLine.getOptionValue("password");
        String database = commandLine.getOptionValue("database");

        Mysql s = new Mysql(cfgFile);
        s.syn(host, Integer.valueOf(port), destination, username, password, database);
    }

    private void parseEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }

            EventType eventType = rowChage.getEventType();
            CanalEntry.Header header = entry.getHeader();
            String tableName = header.getTableName();

            if (!StringUtils.equals(this.database, header.getSchemaName())) {
                return;
            }
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    header.getLogfileName(), header.getLogfileOffset(),
                    header.getSchemaName(), header.getTableName(),
                    eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.INSERT) {
                    insert(tableName, rowData.getAfterColumnsList());
                } else if (eventType == EventType.UPDATE) {
//                    delete(tableName, rowData.getBeforeColumnsList());
//                    insert(tableName, rowData.getAfterColumnsList());
                    update(tableName, rowData.getBeforeColumnsList(), rowData.getAfterColumnsList());
                } else if (eventType == EventType.DELETE) {
                    delete(tableName, rowData.getBeforeColumnsList());
                } else {
                }
            }
        }
    }

    // 目前支持的vid是字符串，其他字段包括整型和字符串
    private void insert(String tableName, List<Column> columns) {
        String nebulaModel = "";
        Boolean isTag = false;
        if (tableName.startsWith("tag_")) {
            isTag = true;
        } else if (tableName.startsWith("edge_")) {
            isTag = false;
        } else {
            return;
        }

        List<Map<String, String>> preMap = this.propMap.get(tableName);
        if (preMap == null) return;
        Map<String, Column> columnMap = columns.stream().collect(Collectors.toMap(Column::getName, a -> a,(k1,k2)->k1));
        for (int i = 0; i < preMap.size(); i++) {
            Map<String, Object> map = new TreeMap<String, Object>();
            Map<String, String> prop = preMap.get(i);

            if (isTag) {
                nebulaModel = prop.get("NEBULA_TAG_NAME");
            } else {
                nebulaModel = prop.get("NEBULA_EDGE_NAME");
            }

            for (String key : prop.keySet()) {
                if (StringUtils.equals(key, "NEBULA_TAG_NAME") || StringUtils.equals(key, "NEBULA_EDGE_NAME")) {
                    continue;
                }

                Column column = columnMap.get(key);
                if (column == null) {
                    continue;
                }
                switch (column.getSqlType()) {
                    case -6://smallint
                    case 5://smallint
                    case -5://bigint
                    case 4://整型
                        map.put(prop.get(column.getName()), StringUtils.isEmpty(column.getValue()) ? 0 : Integer.valueOf(column.getValue()));
                        break;
                    case 6://浮点型
                    case 8://浮点型
                        map.put(prop.get(column.getName()), StringUtils.isEmpty(column.getValue()) ? 0 : Double.valueOf(column.getValue()));
                        break;
                    case 3://DECIMAL
                        map.put(prop.get(column.getName()), StringUtils.isEmpty(column.getValue()) ? 0 : new BigDecimal(column.getValue()));
                        break;
                    case 1://字符串
                    case 12://字符串
                    case 2005://字符串
                        map.put(prop.get(column.getName()), "\"" + StringUtils.replace(column.getValue(), "\"", "\\\"") + "\"");
                        break;
                    default:
                        throw new RuntimeException("不支持该类型:" + column.getSqlType());
                }
            }

            String VID = null;
            List<String> keys = new ArrayList<String>();
            List<Object> values = new ArrayList<Object>();
            for (String key : map.keySet()) {
                if (!isTag && (key.equals("source") || key.equals("target"))) {
                    continue;
                }
                Object value = map.get(key);
                if (key.equals("vertex")) {
                    VID = (String)value;
                    continue;
                }
                keys.add(key);
                values.add(value);
            }

            String id = "";
            if (VID != null) {
                id = VID;
            } else {
                id = map.get("source") + " -> " + map.get("target");
            }

            exec(String.format("INSERT %s %s (%s) VALUES %s :(%s);", isTag ? "VERTEX" : "EDGE", nebulaModel, StringUtils.join(keys, ","), id, StringUtils.join(values, ",")));
        }
    }

    // 目前支持的vid是字符串，其他字段包括整型和字符串
    private void update(String tableName, List<Column> beforecolumns, List<Column> columns) {
        String nebulaModel = "";
        Boolean isTag = false;
        if (tableName.startsWith("tag_")) {
            insert(tableName, columns);
            return;
        } else if (tableName.startsWith("edge_")) {
            delete(tableName, beforecolumns);
            insert(tableName, columns);
            return;
        } else {
            return;
        }
    }

    // 目前支持的vid是字符串
    private void delete(String tableName, List<Column> columns) {
        Boolean isTag = false;
        String nebulaModel = "";
        if (tableName.startsWith("tag_")) {
            isTag = true;
        } else if (tableName.startsWith("edge_")) {
            isTag = false;
        } else {
            return;
        }

        List<Map<String, String>> preMap = this.propMap.get(tableName);
        if (preMap == null) return;
        Map<String, Column> columnMap = columns.stream().collect(Collectors.toMap(Column::getName, a -> a,(k1,k2)->k1));
        for (int i = 0; i < preMap.size(); i++) {
            Map<String, String> prop = preMap.get(i);
            Map<String, String> map = new TreeMap<String, String>();
            String VID = null;

            if (isTag) {
                nebulaModel = prop.get("NEBULA_TAG_NAME");
            } else {
                nebulaModel = prop.get("NEBULA_EDGE_NAME");
            }

            for (String key : columnMap.keySet()) {
                if (StringUtils.equals(key, "NEBULA_TAG_NAME") || StringUtils.equals(key, "NEBULA_EDGE_NAME")) {
                    continue;
                }

                Column column = columnMap.get(key);
                if (column == null) {
                    continue;
                }
                if (prop.get(column.getName()) != null && prop.get(column.getName()).equals("vertex")) {
                    VID = "\"" + column.getValue() + "\"";
                    break;
                } else if (prop.get(column.getName()) != null) {
                    if (prop.get(column.getName()).equals("source")) {
                        map.put("source", "\"" + column.getValue() + "\"");
                    } else if (prop.get(column.getName()).equals("target")) {
                        map.put("target", "\"" + column.getValue() + "\"");
                    }
                }
            }

            String id = "";
            if (VID != null) {
                id = VID;
            } else {
                id = map.get("source") + " -> " + map.get("target");
            }

            exec(String.format("DELETE %s %s %s;", isTag ? "VERTEX" : "EDGE", isTag ? "" : nebulaModel, id));
        }
    }

    private void exec(String nGQL) {
        System.out.println(LocalDateTime.now() + ":" + nGQL);
        NebulaPool pool = new NebulaPool();
        Session session = null;
        try {
            NebulaPoolConfig nebulaPoolConfig = new NebulaPoolConfig();
            nebulaPoolConfig.setMaxConnSize(100);

            Config nebula = this.config.getConfig("nebula");
            String graph = nebula.getConfig("address").getStringList("graph").get(0);
            List<HostAddress> addresses = Arrays.asList(new HostAddress(graph.split(":")[0], Integer.valueOf(graph.split(":")[1])));
            pool.init(addresses, nebulaPoolConfig);
            session = pool.getSession(nebula.getString("user"), nebula.getString("pswd"), false);
            //create schema
            nGQL = "USE " + nebula.getString("space") + ";" + nGQL;
            ResultSet resp = session.execute(nGQL);
            printResult(resp);
        } catch (UnknownHostException e) {
            e.printStackTrace();
        } catch (AuthFailedException e) {
            e.printStackTrace();
        } catch (NotValidConnectionException e) {
            e.printStackTrace();
        } catch (IOErrorException e) {
            e.printStackTrace();
        } finally {
            if (session != null) {
                session.release();
            }
            pool.close();
        }
    }

    private static void printResult(ResultSet resultSet) {
        System.out.println("errorCode: " + resultSet.getErrorCode() + "; errorMsg: " + resultSet.getErrorMessage());
    }
}

/*
*
public final static int BIT = -7;
public final static int TINYINT = -6;
public final static int SMALLINT = 5;
public final static int INTEGER = 4;
public final static int BIGINT = -5;
public final static int FLOAT = 6;
public final static int REAL = 7;
public final static int DOUBLE = 8;
public final static int NUMERIC = 2;
public final static int DECIMAL = 3;
public final static int CHAR = 1;
public final static int VARCHAR = 12;
public final static int LONGVARCHAR = -1;
public final static int DATE = 91;
public final static int TIME = 92;
public final static int TIMESTAMP = 93;
public final static int BINARY = -2;
public final static int VARBINARY = -3;
public final static int LONGVARBINARY = -4;
public final static int NULL = 0;
public final static int OTHER = 1111;
public final static int JAVA_OBJECT = 2000;
public final static int DISTINCT = 2001;
public final static int STRUCT = 2002;
public final static int ARRAY = 2003;
public final static int BLOB = 2004;
public final static int CLOB = 2005;
public final static int REF = 2006;
public final static int DATALINK = 70;
public final static int BOOLEAN = 16;
public final static int ROWID = -8;
public static final int NCHAR = -15;
public static final int NVARCHAR = -9;
public static final int LONGNVARCHAR = -16;
public static final int NCLOB = 2011;
public static final int SQLXML = 2009;

* */